package com.xizi.发布确认;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import com.xizi.utils.RabbitUtils;
import org.junit.Test;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

//发布确认
public class ConfirmMessage {

    public static void main(String[] args) throws Exception {

        ConfirmMessage.publishMessageIndividually(); // 发布1000个单独确认消息,耗时50798ms
        ConfirmMessage.publishMessageBatch(); //发布1000个批量确认消息,耗时1879ms
        ConfirmMessage.publishMessageAsync(); //发布1000个异步确认消息,耗时51ms
    }

    private static final int MESSAGE_COUNT = 1000;

    //1. 单个确认
    public static void publishMessageIndividually() throws Exception {
        try (Channel channel = RabbitUtils.getConnection().createChannel()) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            //开始时间
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                //单个消息就马上进行发布确认
                channel.basicPublish("", queueName, null, message.getBytes());
                //服务端返回 false 或超时时间内未返回，生产者可以消息重发
                boolean flag = channel.waitForConfirms();
//                if(flag){
//                    System.out.println("消息发送成功");
//                }
            }
            //结束时间
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms");
        }
    }


    //2. 批量确认
    public static void publishMessageBatch() throws Exception {
        try (Channel channel = RabbitUtils.getConnection().createChannel())
        {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            //批量确认消息大小
            int batchSize = 100;
            //未确认消息个数
            int outstandingMessageCount = 0;
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                channel.basicPublish("", queueName, null, message.getBytes());
                outstandingMessageCount++;
                if (outstandingMessageCount == batchSize) {
                    channel.waitForConfirms(); //到达100调用一次等待确认
                    outstandingMessageCount = 0;    //重置
                }
            }
            //为了确保还有剩余没有确认消息 再次确认
            if (outstandingMessageCount > 0) {
                channel.waitForConfirms();
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms");
        }
    }

    //3. 异步确认发布
    public static void publishMessageAsync() throws Exception {
        try (Channel channel = RabbitUtils.getConnection().createChannel()) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             * 1.轻松的将序号与消息进行关联
             * 2.轻松批量删除条目 只要给到序列号
             * 3.支持并发访问
             */
            //跳表
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
            // 确认收到消息的一个回调
            // 消息确认成功 回调的函数
            ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
                System.out.println("确认的消息"+deliveryTag);
                //删除确认的消息
                if (multiple) { //1. 批量删除
                    //返回的是小于等于当前序列号的未确认消息 是一个 map
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);
                    //清除该部分未确认消息
                    confirmed.clear();
                }else{ //2. 单个删除
                    //只清除当前序列号的消息
                    outstandingConfirms.remove(deliveryTag);
                }
            };

            //消息确认失败 回调的函数
            ConfirmCallback nackCallback = (deliveryTag, multiple) ->
            {
                String message = outstandingConfirms.get(deliveryTag);
                System.out.println("发布的消息"+message+"未被确认，序列号"+deliveryTag);
            };

            //  添加一个异步确认的监听器 监听哪些成功了 哪些失败了
            // 1.确认收到消息的回调
            // 2.未收到消息的回调
            channel.addConfirmListener(ackCallback, nackCallback);

            //开启时间
            long begin = System.currentTimeMillis();
            //批量发送消息
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = "消息" + i;
                /**
                 * channel.getNextPublishSeqNo()获取下一个消息的序列号
                 * 通过序列号与消息体进行一个关联
                 * 全部都是未确认的消息体
                 */
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", queueName, null, message.getBytes());
            }
            // 结束时间
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) +"ms");
        }
    }
}
